package com.zdd.ehe.controller;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.zdd.ehe.EheApplication;
import com.zdd.ehe.entity.DiseasePredict;
import com.zdd.ehe.result.Result;
import com.zdd.ehe.service.DiseasePredictService;
import com.zdd.ehe.service.KafkaService;
import com.zdd.ehe.vo.PredictVo;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

/**
 * @version 1.0.0
 * @title: PredictContorller
 * @author: HFLy
 * @description:
 * @date: 7/3/2023 下午3:29
 */
@RestController
@RequestMapping("/predict")
@Api(tags = "图像识别")
public class PredictContorller {
    public static final String USER="root";
    public static final String URI="hdfs://192.168.111.111:9000";

    public static final String ROOT_PATH ="/data/images/";
    @Autowired
    public KafkaService kafkaService;
    @Autowired
    public DiseasePredictService diseasePredictService;
    @ApiOperation("图像识别")
    @PostMapping("/image")
    public Result image(@RequestParam("image") @RequestPart MultipartFile image) throws IOException {
        System.out.println("图像识别");
        System.setProperty("HADOOP_USER_NAME",USER);
        // 将文件上传到文件系统
        InputStream inputStream = null;
        FSDataOutputStream fsDataOutputStream = null;
        String timeStr =String.valueOf(new Date().getTime()) ;
        String pathStr= ROOT_PATH+ timeStr +".jpg";
        System.out.println(pathStr);
        try {
            inputStream = image.getInputStream();FileSystem hdfs = FileSystem.get(new URI(URI), new Configuration(),USER);
            Path desPath = new Path(pathStr);
            fsDataOutputStream = hdfs.create(desPath);
            byte[] b = new byte[4096];
            int read;
            while ((read=inputStream.read(b)) > 0) {
                fsDataOutputStream.write(b, 0, read);
            }
            fsDataOutputStream.flush();
        } catch (IOException | URISyntaxException | InterruptedException exception) {
            exception.printStackTrace();
        } finally {
            if (null != inputStream) inputStream.close();
            if (null != fsDataOutputStream) fsDataOutputStream.close();
        }
        EheApplication.FINISH_PREDICT=false;
        //获取锁
        while (!EheApplication.PREDICT_LOCK){
            EheApplication.PREDICT_LOCK = true;
            kafkaService.sendToPredict(pathStr);
            while (!EheApplication.FINISH_PREDICT){
                continue;
            }
            EheApplication.PREDICT_LOCK = false;
            break;
        }
        List<PredictVo> predictVos = JSON.parseArray(EheApplication.PREDICT_RESULT, PredictVo.class);
        ArrayList<String> nameList = new ArrayList<>();
        HashMap<String,Object> map = new HashMap<>();
        for (PredictVo predictVo : predictVos) {
            nameList.add(predictVo.getName());
            map.put(predictVo.getName(),predictVo);
        }
        List<DiseasePredict> diseasePredicts = diseasePredictService.list(new QueryWrapper<DiseasePredict>().in("name", nameList));
        for (DiseasePredict diseasePredict : diseasePredicts) {
            PredictVo predict = (PredictVo)map.get(diseasePredict.getName());
            BeanUtils.copyProperties(diseasePredict,predict);
        }
        return Result.success("识别成功",predictVos);
    }
}
